package com.hmeter.hadoop.hdfs;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Arrays;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.DefaultCodec;

/**
 * Utility for HDFS file operations
 * 
 * @author dino.keco
 * @author bakir.jusufbegovic
 * @version 1.0
 * @since 1.0
 * 
 */
public class HDFSUtil {

	public static final String[] COODECS = new String[] { "org.apache.hadoop.io.compress.BZip2Codec", "org.apache.hadoop.io.compress.GzipCodec" };

	/**
	 * For small files
	 * 
	 * @param conf
	 * @param path
	 * @param content
	 * @param overwrite
	 * @param codec
	 * @throws IOException
	 */
	public static void createFile(Configuration conf, String path, String content, boolean overwrite, String codec) throws IOException {
		FileSystem fs = FileSystem.get(conf);
		OutputStream out = null;

		Path p = new Path(path);

		if (fs.exists(p) && !overwrite) {
			// append 
			if (codec != null && !"".equals(codec)) {
				out = getCodec(conf, codec).createOutputStream(fs.append(p));
			} else {
				out = fs.append(p);
			}
		} else {
			// create
			if (codec != null && !"".equals(codec)) {
				out = getCodec(conf, codec).createOutputStream(fs.create(p, overwrite));
			} else {
				out = fs.create(p, overwrite);
			}
		}
		try {
			out.write(content.getBytes());
			out.flush();
		} finally {
			try {
				out.close();
			} catch (IOException e) {
				// ignore
			}
		}
	}

	/**
	 * For big files
	 * 
	 * @param conf
	 * @param path
	 * @param input
	 * @param overwrite
	 * @param codec
	 * @throws IOException
	 */
	public static void createFile(Configuration conf, String path, InputStream input, boolean overwrite, String codec) throws IOException {
		FileSystem fs = FileSystem.get(conf);
		OutputStream out = null;
		if (codec != null && !"".equals(codec)) {
			out = getCodec(conf, codec).createOutputStream(fs.create(new Path(path), overwrite));
		} else {
			out = fs.create(new Path(path), overwrite);
		}
		try {
			byte[] buffer = new byte[2048];
			int read = 0;
			while ((read = input.read(buffer)) > 0) {
				out.write(buffer, 0, read);
			}
			out.flush();
		} finally {
			try {
				out.close();
			} catch (IOException e) {
				// ignore
			}
		}
	}

	public static void deleteFile(Configuration conf, String path, boolean recursive) throws IOException {
		FileSystem fs = FileSystem.get(conf);
		fs.delete(new Path(path), recursive);
	}

	public static void renameFile(Configuration conf, String oldPath, String newPath) throws IOException {
		FileSystem fs = FileSystem.get(conf);
		fs.rename(new Path(oldPath), new Path(newPath));
	}

	public static FileContent tailFile(Configuration conf, String path, String codec, long offset, int size, boolean info) throws IOException {
		FileSystem fs = FileSystem.get(conf);
		Path p = new Path(path);
		InputStream in = null;
		if (codec != null && !"".equals(codec)) {
			in = getCodec(conf, codec).createInputStream(fs.open(new Path(path)));
		} else {
			in = fs.open(new Path(path));
		}

		try {
			in.skip(offset);

			byte[] content = new byte[(int) size];
			int read = in.read(content);

			FileContent fc = new FileContent();
			fc.setPath(p.toUri().toString());
			if (info) {
				FileStatus fileStatus = fs.getFileStatus(p);
				fc.setSize(String.valueOf(fileStatus.getLen()));
				fc.setBlockSize(String.valueOf(fileStatus.getBlockSize()));
				fc.setReplication(String.valueOf(fileStatus.getReplication()));
				fc.setOwner(fileStatus.getOwner());
				fc.setGroup(fileStatus.getGroup());
			}

			fc.setContent(new String(Arrays.copyOf(content, read)));
			return fc;
		} finally {
			in.close();
		}

	}

	private static CompressionCodec getCodec(Configuration conf, String clazz) throws IOException {
		try {
			@SuppressWarnings("unchecked")
			Class<? extends CompressionCodec> codecClass = (Class<? extends CompressionCodec>) Class.forName(clazz);
			CompressionCodec coder = codecClass.newInstance();
			if (coder instanceof DefaultCodec) {
				((DefaultCodec) coder).setConf(conf);
			}
			return coder;
		} catch (Exception e) {
			throw new IOException("Failed to load codec " + clazz, e);
		}
	}

	public static class FileContent {
		private String content;
		private String path;
		private String size;
		private String blockSize;
		private String replication;
		private String owner;
		private String group;

		public String getContent() {
			return content;
		}

		public void setContent(String content) {
			this.content = content;
		}

		public String getPath() {
			return path;
		}

		public void setPath(String path) {
			this.path = path;
		}

		public String getSize() {
			return size;
		}

		public void setSize(String size) {
			this.size = size;
		}

		public String getBlockSize() {
			return blockSize;
		}

		public void setBlockSize(String blockSize) {
			this.blockSize = blockSize;
		}

		public String getReplication() {
			return replication;
		}

		public void setReplication(String replication) {
			this.replication = replication;
		}

		public String getOwner() {
			return owner;
		}

		public void setOwner(String owner) {
			this.owner = owner;
		}

		public String getGroup() {
			return group;
		}

		public void setGroup(String group) {
			this.group = group;
		}

		public String toXMLString() {
			StringBuilder builder = new StringBuilder();
			builder.append("<file>");

			builder.append("<info>");

			if (path != null)
				builder.append("<path>").append(path).append("</path>");
			if (size != null)
				builder.append("<size>").append(size).append("</size>");
			if (blockSize != null)
				builder.append("<blockSize>").append(blockSize).append("</blockSize>");
			if (replication != null)
				builder.append("<replication>").append(replication).append("</replication>");
			if (owner != null)
				builder.append("<owner>").append(owner).append("</owner>");
			if (group != null)
				builder.append("<group>").append(group).append("</group>");

			builder.append("</info>");

			if (content != null)
				builder.append("<content>").append(content).append("</content>");

			builder.append("</file>");
			return builder.toString();
		}
	}
}
